package com.shujia.flink.core;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class Demo04Proctime {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> wordsDS = env.socketTextStream("master", 8888);

        DataStream<Tuple2<String, Integer>> kvDS = wordsDS.map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT));
        // 每隔5s统计单词的数量
        // 需要使用滚动窗口
//        kvDS.keyBy(kv -> kv.f0)
//                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
//                .sum(1)
//                .print();

        // 每隔5s统计最近10s内的单词数量
        // 需要使用滑动窗口
        // 适用榜单的统计
        // 需要指定两个时间：窗口的大小、滑动的时间
        // 当窗口大小 等于 滑动时间时，可以视为滚动窗口
        // 当窗口大小 大于 滑动时间时，可以视为滑动窗口
        // 当然没有窗口大小 小于 滑动时间 这种情况
        kvDS.keyBy(kv -> kv.f0)
                .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .sum(1)
                .print();

        env.execute();
    }
}
